package com.marveal.rabbit.producer.broker;

import com.marveal.rabbit.api.Message;
import com.marveal.rabbit.api.MessageType;
import com.marveal.rabbit.api.exception.MessageRuntimeException;
import com.marveal.rabbit.api.exception.MessageRuntimeExceptionType;
import com.marveal.rabbit.common.convert.GenericMessageConverter;
import com.marveal.rabbit.common.convert.RabbitMessageConverter;
import com.marveal.rabbit.common.serializer.Serializer;
import com.marveal.rabbit.common.serializer.SerializerFactory;
import com.marveal.rabbit.common.serializer.impl.KryoSerializerFactory;
import com.marveal.rabbit.common.util.Assert;
import com.marveal.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    /**
     * String: topic
     */
    private final Map<String, RabbitTemplate> rabbitTemplateMap = new ConcurrentHashMap<>();

    /**
     * 分割符
     */
    private static final String Splitter = "#";

    private final SerializerFactory serializerFactory = KryoSerializerFactory.INSTANCE;

    @Autowired
    private ConnectionFactory connectionFactory;

    private MessageStoreService messageStoreService;

    private RabbitTemplateContainer() {
    }

    public RabbitTemplateContainer(MessageStoreService messageStoreService) {
        this.messageStoreService = messageStoreService;
    }

    public RabbitTemplate getRabbitTemplate(Message message) throws MessageRuntimeException {
        Assert.notNull(message, MessageRuntimeExceptionType.MESSAGE_IS_NULL.description);
        Assert.notEmpty(message.getTopic(), MessageRuntimeExceptionType.MESSAGE_TOPIC_IS_EMPTY.description);

        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitTemplateMap.get(topic);
        if (rabbitTemplate != null) {
            return rabbitTemplate;
        }
        log.info("#RabbitTemplateContainer.getRabbitTemplate# topic: {} is not exist, create one rabbitTemplate", topic);

        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        newRabbitTemplate.setExchange(topic);
        newRabbitTemplate.setRoutingKey(message.getRoutingKey());
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());

        // 设置message的序列化方式
        Serializer serializer = serializerFactory.create();
        GenericMessageConverter gmc = new GenericMessageConverter(serializer);
        RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
        newRabbitTemplate.setMessageConverter(rmc);

        MessageType messageType = message.getMessageType();
        if (MessageType.RELIANT.type == messageType.type) {
            newRabbitTemplate.setConfirmCallback(this);
        }
        rabbitTemplateMap.putIfAbsent(topic, newRabbitTemplate);
        return rabbitTemplateMap.get(topic);
    }



    /**
     * 具体的消息应答
     *
     * @param correlationData 唯一ID
     * @param ack             确认
     * @param cause           异常
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // check correlationData is not null
        Assert.notNull(correlationData, MessageRuntimeExceptionType.MESSAGE_CORRELATION_DATA_IS_NULL.description);
        String correlationDataId = correlationData.getId();
        Assert.notNull(correlationDataId, MessageRuntimeExceptionType.MESSAGE_CORRELATION_DATA_ID_IS_NULL.description);

        String[] dates = correlationDataId.split(Splitter);
        // check correlationData length is 2 after split
        Assert.notEmpty(dates, MessageRuntimeExceptionType.MESSAGE_CORRELATION_DATA_ID_IS_NULL.description);
        if (dates.length != 3) {
            log.error("#RabbitTemplateContainer.confirm# correlationDataId: {} is deny", correlationDataId);
            throw new MessageRuntimeException(MessageRuntimeExceptionType.MESSAGE_CORRELATION_DATA_ID_ERROR);
        }

        String messageId = dates[0];
        Long timestamp = Long.parseLong(dates[1]);
        int messageType = Integer.parseInt(dates[2]);

        if (ack) {
            if (MessageType.RELIANT.type == messageType) {
                // set database message status 1
                messageStoreService.success(messageId);
            }
            log.info("send message is OK, confirm messageId: {}, sendTime: {}", messageId, timestamp);
        } else {
            log.error("send message is failed, confirm messageId: {}, sendTime: {}", messageId, timestamp);
        }


    }
}
